package com.ultrapower.hbase;

import java.io.IOException;
import java.nio.charset.Charset;

import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;

import com.sematext.hbase.wd.AbstractRowKeyDistributor;
import com.sematext.hbase.wd.DistributedScanner;
/**
 * 
 * @author wangy
 *
 */
public class HbaseTemplate {
	private static final Charset CHARSET = Charset.forName("UTF-8");
	protected static final int MAX_SCAN_CACHE_SIZE = 256;
	private HbaseConnectionFactory tableFactory;

	public Charset getCharset() {
		return CHARSET;
	}

	public <T> T find(TableName tableName, Scan scan, ResultsExtractor<T> action) {
		return execute(tableName, scan, action);
	}
	public <T> T find(TableName tableName, String family, String qualifier, ResultsExtractor<T> action) {
		Scan scan = new Scan();
		scan.addColumn(family.getBytes(getCharset()), qualifier.getBytes(getCharset()));
		return execute(tableName, scan, action);
	}

	public <T> T find(TableName tableName, Scan scan, AbstractRowKeyDistributor rowKeyDistributor,final ResultsExtractor<T> action) {
		Table table = tableFactory.getTable(tableName);
		try {
			final ResultScanner[] splitScanners = splitScan(table, scan, rowKeyDistributor);
            final ResultScanner scanner = new DistributedScanner(rowKeyDistributor, splitScanners);
			return action.extractData(scanner);
		} catch (Throwable e) {
			throw new RuntimeException(e);
		} finally {
			tableFactory.releaseTable(table);
		}
	}
	public <T> T find(TableName tableName, String family, final ResultsExtractor<T> action) {
		Scan scan = new Scan();
		scan.addFamily(family.getBytes(getCharset()));
		return execute(tableName, scan, action);
	}

	private <T> T execute(TableName tableName, final Scan scan, ResultsExtractor<T> action) {
		Table table = tableFactory.getTable(tableName);
		try {
			final ResultScanner scanner = table.getScanner(scan);
			return action.extractData(scanner);
		} catch (Throwable e) {
			throw new RuntimeException(e);
		} finally {
			tableFactory.releaseTable(table);
		}
	}

	public <T> T get(TableName tableName, final Get get, final ResultExtractor<T> action) {
		Table table = tableFactory.getTable(tableName);
		try {
			Result result = table.get(get);
			return action.extractData(result);
		} catch (Throwable e) {
			throw new RuntimeException(e);
		} finally {
			tableFactory.releaseTable(table);
		}
    }
	public <T> T get(TableName tableName, final byte[] rowName, final byte[] familyName, final byte[] qualifier,
			final ResultExtractor<T> action) {
		Get get = new Get(rowName);
		if (familyName != null) {
			if (qualifier != null) {
				get.addColumn(familyName, qualifier);
			} else {
				get.addFamily(familyName);
			}
		}
		Table table = tableFactory.getTable(tableName);
		try {
			Result result = table.get(get);
			return action.extractData(result);
		} catch (Throwable e) {
			throw new RuntimeException(e);
		} finally {
			tableFactory.releaseTable(table);
		}
	}
	 private ResultScanner[] splitScan(Table table, Scan originalScan, AbstractRowKeyDistributor rowKeyDistributor) throws IOException {
	        Scan[] scans = rowKeyDistributor.getDistributedScans(originalScan);
	        final int length = scans.length;
	        for (int i = 0; i < length; i++) {
	            Scan scan = scans[i];
	            // other properties are already set upon construction
	            scan.setId(scan.getId() + "-" + i);
	        }

	        ResultScanner[] scanners = new ResultScanner[length];
	        boolean success = false;
	        try {
	            for (int i = 0; i < length; i++) {
	                scanners[i] = table.getScanner(scans[i]);
	            }
	            success = true;
	        } finally {
	            if (!success) {
	                closeScanner(scanners);
	            }
	        }
	        return scanners;
	    }
	 
	 private void closeScanner(ResultScanner[] scannerList) {
	        for (ResultScanner scanner : scannerList) {
	            if (scanner != null) {
	                try {
	                    scanner.close();
	                } catch (Exception e) {
	                	e.printStackTrace();
	                    System.out.println("Scanner.close() error Caused:{}"+ e.getMessage() + e);
	                }
	            }
	        }
	    }
	public HbaseConnectionFactory getTableFactory() {
		return tableFactory;
	}

	public void setTableFactory(HbaseConnectionFactory tableFactory) {
		this.tableFactory = tableFactory;
	}

}
